import time
import threading
import queue
import pathway as pw
from typing import Any, Dict, List, Union


class BaseBatchObserver(pw.io.python.ConnectorObserver):
    """
    批量数据观察者基类，使用线程安全队列进行缓冲，提供定时与批处理逻辑。
    每 1 秒检查一次缓冲区，并在达到 batch_size 或时间窗口触发批处理。

    :param batch_size: 达到该条数时触发批处理。
    :param time_window: 上次批处理后经过该秒数触发批处理（秒）。
    """

    def __init__(self, batch_size=20, time_window=60):
        super().__init__()
        self.batch_size = batch_size
        self.time_window = time_window
        self._stop_event = threading.Event()
        self.batch_queue = queue.Queue()  # 线程安全队列
        # 创建守护线程
        self.thread = threading.Thread(target=self._run, daemon=True)
        self.thread.start()

    def on_change(self, key: pw.Pointer, row: dict, time: int, is_addition: bool):
        if row:
            self.batch_queue.put(row)  # 入队
        if self.batch_queue.qsize() >= self.batch_size:
            self._process_queue_batch()

    def on_end(self):
        self._stop_event.set()
        self.thread.join()
        self._process_queue_batch()  # 处理剩余数据

    def _run(self):
        """
        守护线程执行逻辑，循环调用定时任务方法
        """
        while not self._stop_event.is_set():
            start_time = time.time()

            self._process_queue_batch()

            # 计算等待时间，保证固定周期运行
            elapsed = time.time() - start_time
            time_to_wait = max(0, self.time_window - elapsed)
            if self._stop_event.wait(timeout=time_to_wait):
                break

    def _process_queue_batch(self):
        """
        从队列中一次性取出所有数据进行批处理
        """
        batch_data = []
        while not self.batch_queue.empty():
            try:
                batch_data.append(self.batch_queue.get_nowait())
            except queue.Empty:
                break

        if batch_data:
            self.process_batch(batch_data)

    def convert_to_dict(self, obj: Any) -> Union[Dict, List, Any]:
        """
        将 Pathway 的 Json 对象转换为 Python 字典
        
        :param obj: 需要转换的对象
        :return: 转换后的Python对象（字典、列表或其他类型）
        """
        # 如果是Pathway对象且有as_dict方法，使用该方法转换
        if hasattr(obj, 'as_dict'):
            return obj.as_dict()
        
        # 如果是字典类型，直接返回
        if isinstance(obj, dict):
            return obj
            
        # 如果是列表类型，递归处理列表中的每个元素
        if isinstance(obj, list):
            return [self.convert_to_dict(item) for item in obj]
            
        # 如果是其他类型，尝试用dict()转换
        try:
            return dict(obj)
        except (TypeError, ValueError):
            # 如果无法转换为字典，返回原对象
            return obj

    def process_batch(self, batch_data):
        """
        子类需实现该方法，定义具体的批处理逻辑（在该方法中可以进行耗时操作）。

        :param batch_data: 本次需要处理的批量数据列表。
        """
        raise NotImplementedError("子类必须实现 process_batch 方法")